#!/usr/bin/env python3
"""
GStreamer Editing Services MCP Server

面向软件：gst-editing-services (ges-launch-1.0)

MCP tools 列表：
1. ges_launch_clip - 添加视频/音频剪辑片段
   参数：file_path, inpoint, duration, start, track_types, layer, alpha
   返回：执行结果 JSON

2. ges_launch_effect - 为剪辑添加特效
   参数：file_path, effect_name, set_dusts, alpha
   返回：执行结果 JSON

3. ges_launch_test_clip - 生成测试剪辑
   参数：pattern, name, start, duration, inpoint, layer
   返回：执行结果 JSON

4. ges_launch_title - 创建文本标题
   参数：text, start, duration, inpoint, layer
   返回：执行结果 JSON

5. ges_launch_set_properties - 设置剪辑属性
   参数：file_path, properties (字典，如 {'alpha': 0.5, 'volume': 0.8})
   返回：执行结果 JSON
"""

import subprocess
import json
from typing import List, Dict, Optional
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("GStreamer Editing Services MCP Server")


def run_ges_command(command_parts: List[str]) -> dict:
    """执行 ges-launch-1.0 命令并返回统一格式的结果"""
    try:
        command_str = " ".join(command_parts)
        result = subprocess.run(
            command_parts,
            capture_output=True,
            text=True,
            timeout=30
        )
        
        return {
            "success": result.returncode == 0,
            "command": command_str,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except subprocess.TimeoutExpired:
        return {
            "success": False,
            "command": " ".join(command_parts),
            "exit_code": -1,
            "stdout": "",
            "stderr": "Command timed out after 30 seconds"
        }
    except Exception as e:
        return {
            "success": False,
            "command": " ".join(command_parts),
            "exit_code": -1,
            "stdout": "",
            "stderr": f"Command execution failed: {str(e)}"
        }


@mcp.tool()
def ges_launch_clip(
    file_path: str,
    inpoint: Optional[float] = None,
    duration: Optional[float] = None,
    start: Optional[float] = None,
    track_types: Optional[str] = None,
    layer: Optional[int] = None,
    alpha: Optional[float] = None
) -> dict:
    """
    添加视频/音频剪辑片段到时间线
    
    参数:
        file_path: 媒体文件路径
        inpoint: 剪辑入点时间（秒）
        duration: 剪辑持续时间（秒）
        start: 在时间线中的开始时间（秒）
        track_types: 轨道类型（如 'audio'）
        layer: 图层编号
        alpha: 透明度值（0.0-1.0）
    
    返回:
        JSON结构: {"success": bool, "command": str, "exit_code": int, "stdout": str, "stderr": str}
    """
    command_parts = ["ges-launch-1.0"]
    
    if track_types:
        command_parts.extend(["--track-types", track_types])
    
    command_parts.extend(["+clip", file_path])
    
    if inpoint is not None:
        command_parts.append(f"inpoint={inpoint}")
    
    if duration is not None:
        command_parts.append(f"duration={duration}")
    
    if start is not None:
        command_parts.append(f"start={start}")
    
    if layer is not None:
        command_parts.append(f"layer={layer}")
    
    if alpha is not None:
        command_parts.extend(["set-alpha", str(alpha)])
    
    return run_ges_command(command_parts)


@mcp.tool()
def ges_launch_effect(
    file_path: str,
    effect_name: str,
    set_dusts: Optional[bool] = None,
    alpha: Optional[float] = None
) -> dict:
    """
    为剪辑片段添加特效
    
    参数:
        file_path: 媒体文件路径
        effect_name: 特效名称（如 'agingtv'）
        set_dusts: 是否设置灰尘效果
        alpha: 透明度值（0.0-1.0）
    
    返回:
        JSON结构: {"success": bool, "command": str, "exit_code": int, "stdout": str, "stderr": str}
    """
    command_parts = ["ges-launch-1.0", "+clip", file_path, "+effect", effect_name]
    
    if set_dusts is not None:
        command_parts.extend(["set-dusts", "true" if set_dusts else "false"])
    
    if alpha is not None:
        command_parts.extend(["set-alpha", str(alpha)])
    
    return run_ges_command(command_parts)


@mcp.tool()
def ges_launch_test_clip(
    pattern: str,
    name: Optional[str] = None,
    start: Optional[float] = None,
    duration: Optional[float] = None,
    inpoint: Optional[float] = None,
    layer: Optional[int] = None
) -> dict:
    """
    生成测试剪辑模式
    
    参数:
        pattern: 测试模式名称（如 'pattern1', 'pattern2' 等）
        name: 剪辑名称
        start: 开始时间（秒）
        duration: 持续时间（秒）
        inpoint: 入点时间（秒）
        layer: 图层编号
    
    返回:
        JSON结构: {"success": bool, "command": str, "exit_code": int, "stdout": str, "stderr": str}
    """
    command_parts = ["ges-launch-1.0", "+test-clip", pattern]
    
    if name is not None:
        command_parts.append(f"name={name}")
    
    if start is not None:
        command_parts.append(f"start={start}")
    
    if duration is not None:
        command_parts.append(f"duration={duration}")
    
    if inpoint is not None:
        command_parts.append(f"inpoint={inpoint}")
    
    if layer is not None:
        command_parts.append(f"layer={layer}")
    
    return run_ges_command(command_parts)


@mcp.tool()
def ges_launch_title(
    text: str,
    start: Optional[float] = None,
    duration: Optional[float] = None,
    inpoint: Optional[float] = None,
    layer: Optional[int] = None
) -> dict:
    """
    创建文本标题剪辑
    
    参数:
        text: 标题文本内容
        start: 开始时间（秒）
        duration: 持续时间（秒）
        inpoint: 入点时间（秒）
        layer: 图层编号
    
    返回:
        JSON结构: {"success": bool, "command": str, "exit_code": int, "stdout": str, "stderr": str}
    """
    command_parts = ["ges-launch-1.0", "+title", text]
    
    if start is not None:
        command_parts.append(f"start={start}")
    
    if duration is not None:
        command_parts.append(f"duration={duration}")
    
    if inpoint is not None:
        command_parts.append(f"inpoint={inpoint}")
    
    if layer is not None:
        command_parts.append(f"layer={layer}")
    
    return run_ges_command(command_parts)


@mcp.tool()
def ges_launch_set_properties(
    file_path: str,
    properties: Dict[str, float]
) -> dict:
    """
    设置剪辑属性
    
    参数:
        file_path: 媒体文件路径
        properties: 属性字典，如 {'alpha': 0.5, 'volume': 0.8}
    
    返回:
        JSON结构: {"success": bool, "command": str, "exit_code": int, "stdout": str, "stderr": str}
    """
    command_parts = ["ges-launch-1.0", "+clip", file_path]
    
    for prop_name, prop_value in properties.items():
        command_parts.extend([f"set-{prop_name}", str(prop_value)])
    
    return run_ges_command(command_parts)


if __name__ == "__main__":
    mcp.run()